package org.stayfool.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.Test;

public class NIOSocketServer {

	private Selector s;

	private AtomicReference<Selector> ss = new AtomicReference<Selector>(null);

	private AtomicBoolean started = new AtomicBoolean(false);

	private ExecutorService es = Executors.newFixedThreadPool(8);

	public static void main(String[] args) throws Exception {
		NIOSocketServer server = new NIOSocketServer();
		server.start();
	}

	@Test
	public void start() {
		try {
			initHandle();
			initReceive();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void initReceive() {
		es.execute(() -> {
			try {
				s = Selector.open();
				ss.set(Selector.open());
				ServerSocketChannel sc = ServerSocketChannel.open();
				sc.socket().bind(new InetSocketAddress(9999));
				sc.configureBlocking(false);

				sc.register(s, SelectionKey.OP_ACCEPT);

				started.set(true);
				System.out.println("server started : " + LocalDateTime.now().toString());

				// receive
				while (true) {
					int readyChannels = 0;
					try {
						readyChannels = s.select();
					} catch (IOException e) {
						e.printStackTrace();
					}
					if (readyChannels == 0)
						continue;
					Set<SelectionKey> selectedKeys = s.selectedKeys();
					Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
					while (keyIterator.hasNext()) {
						SelectionKey key = keyIterator.next();
						if (key.isAcceptable()) {
							// a connection was accepted by a ServerSocketChannel.

							System.out.println(" accepted ");

							ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
							SocketChannel c = ssc.accept();
							// c.configureBlocking(false);
							// c.register(ss.get(), SelectionKey.OP_READ | SelectionKey.OP_WRITE);

							ByteBuffer wel = ByteBuffer.allocate(128);
							wel.put("welcome".getBytes());
							wel.flip();

							try {
								c.write(wel);
							} catch (IOException e) {
								e.printStackTrace();
							}

							System.out.println(" distribute ");
						}
						keyIterator.remove();
					}
				}

			} catch (IOException e) {
				e.printStackTrace();
			}

		});
	}

	public void initHandle() {

		es.execute(() -> {
			// wait for server start
			while (!started.get()) {
				System.out.println("wait for server start");
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// receive
			while (true) {
				System.out.println("wait for client connect ");
				int readyChannels = 0;
				try {
					readyChannels = ss.get().select();
				} catch (IOException e) {
					e.printStackTrace();
				}
				if (readyChannels == 0)
					continue;
				System.out.println("client connected ");
				Set<SelectionKey> selectedKeys = ss.get().selectedKeys();
				Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
				while (keyIterator.hasNext()) {
					SelectionKey key = keyIterator.next();
					if (key.isAcceptable()) {
						// a connection was accepted by a ServerSocketChannel.
						System.out.println(" never reachable ");
					}
					if (key.isConnectable()) {
						// a connection was established with a remote server.
						System.out.println(" never reachable ");
						ByteBuffer wel = ByteBuffer.allocate(128);
						wel.put("welcome".getBytes());
						wel.flip();
						SocketChannel c = (SocketChannel) key.channel();

						try {
							c.write(wel);
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					if (key.isReadable()) {
						// a channel is ready for reading
						System.out.println(" read ");
						ByteBuffer wel = ByteBuffer.allocate(128);
						SocketChannel c = (SocketChannel) key.channel();
						try {
							c.read(wel);
							wel.flip();
							while (wel.hasRemaining()) {
								System.out.println((char) wel.get());
							}
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					if (key.isWritable()) {
						// a channel is ready for writing
						System.out.println(" write ");
						ByteBuffer wel = ByteBuffer.allocate(128);
						wel.put("welcome".getBytes());
						wel.flip();
						SocketChannel c = (SocketChannel) key.channel();

						try {
							c.write(wel);
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					keyIterator.remove();
				}
			}

		});
	}
}
